/** @file    Scheduler.cpp
 *  @author  Alessandro Polo
 *  @version $Id: Scheduler.cpp 2891 2010-08-12 01:58:59Z alex $
 *  @brief
 * File containing methods for the wosh::xtended::Scheduler class.
 * The header for this class can be found in Scheduler.h, check that file
 * for class description.
 ****************************************************************************/
/* Copyright (c) 2007-2010, WOSH - Wide Open Smart Home 
 * by Alessandro Polo - OpenSmartHome.com
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *     * Redistributions of source code must retain the above copyright
 *       notice, this list of conditions and the following disclaimer.
 *     * Redistributions in binary form must reproduce the above copyright
 *       notice, this list of conditions and the following disclaimer in the
 *       documentation and/or other materials provided with the distribution.
 *     * Neither the name of the OpenSmartHome.com WOSH nor the
 *       names of its contributors may be used to endorse or promote products
 *       derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY Alessandro Polo ''AS IS'' AND ANY
 * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL Alessandro Polo BE LIABLE FOR ANY
 * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 ****************************************************************************/

 #include "Scheduler.h"

 #include <core/ObjectAllocator.h>
 #include <core/DataModels.h>

 #include <time.h>


using namespace std;
namespace wosh {
 namespace xtended {

bool SchedulerTaskSortPredicate(const SchedulerTask* d1, const SchedulerTask* d2) {
	if ( d1 == NULL ) return true;
	if ( d2 == NULL ) return false;
	if ( d1->getTimeTick() < d2->getTimeTick() ) return true;
	if ( d1->getTimeTick() > d2->getTimeTick() ) return false;
	return d1->getPriority() < d2->getPriority();
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

 WOSH_REGISTER(wosh::xtended::Scheduler, "wosh::Thread", _Scheduler_VERSION, _static_Scheduler )

//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS

Scheduler::Scheduler()
	: ThreadImpl(NULL, "Scheduler.thread") {

	this->listener = NULL;
	this->disabled = false;

	this->workersPool = new ThreadPool<SchedulerTaskExecuter>();

	Log.setIndent( 4 );
	Log.setContext( "Scheduler" );
	Log.setLevel( LOG_INFO );
 }

Scheduler::~Scheduler() {
	delete this->workersPool; this->workersPool = NULL;
 }

//////////////////////////////////////////////////////////////////////////////////////////////// CONSTRUCTORS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////// FLOW CONTROL

WRESULT Scheduler::startScheduler( unsigned long synch_timeout ) {
	Log(LOG_INFO, ":startScheduler(%lu) Starting Thread(s)..", synch_timeout );
	WRESULT ret = ThreadImpl::startThread(synch_timeout);
	Log(LOG_VERBOSE, ":startScheduler(%lu) Starting Workers..", synch_timeout );
	this->workersPool->waitApplyState(Thread::STATE_RUNNING, synch_timeout);
	return ret;
 }

WRESULT Scheduler::stopScheduler( unsigned long synch_timeout ) {
	Log(LOG_INFO, ":stopScheduler(%lu) Stopping Thread(s)..", synch_timeout );
	WRESULT ret = ThreadImpl::quitThread(synch_timeout);
	Log(LOG_VERBOSE, ":stopScheduler(%lu) Stopping Workers..", synch_timeout );
	this->workersPool->waitApplyState(Thread::STATE_STOPPED, synch_timeout);
	return ret;
 }

WRESULT Scheduler::setThreadPoolCount( int worker_threads ) {
	return this->workersPool->setWorkersCount<Scheduler>(worker_threads, this);
 }

//////////////////////////////////////////////////////////////////////////////////////////////// FLOW CONTROL
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD

bool Scheduler::initializingThread() {
	if ( this->listener == NULL ) {
		return false;
	 }
	return true;
 }

void Scheduler::runThread() {
	Log(LOG_DEBUG, ":runThread(): STARTING" );

	while(this->running) {
		setThreadAlive();
		this->configMux.lockForRead();
		if ( this->disabled ) {
			this->configMux.unLock();
			ThreadImpl::sleepForMSec(300);
			continue;
		 }
		this->configMux.unLock();
		this->tasks.transactionBeginWrite();
		if ( this->tasks.size() == 0 ) {
			this->tasks.transactionEnd();
			ThreadImpl::sleepForMSec(125);
			continue;
		 }
		// get first item and remove it from queue, it's always the next one to be executed
		SchedulerTask* task = this->tasks.frontItem();
		if ( task == NULL ) {
			// fix the list and restart loop
			this->tasks.erase(this->tasks.begin(), false);
			this->tasks.transactionEnd();
			ThreadImpl::sleepForMSec(125);
			continue;
		 }
		int64 now = currentCronTime();
		// eval if we are near to exec it
		if ( task->getTimeTick() - now < 0 ) { // we are late!
			Log(LOG_WARNING, " Task#%"PRId64" : POST-PONED EXECUTION (%"PRId64"-%"PRId64"=%"PRId64"s late)!", task->getID(), now, task->getTimeTick(), now-task->getTimeTick() );
		 }
		else if ( task->getTimeTick() - now > 2 ) { // seconds
			this->tasks.transactionEnd();
			ThreadImpl::sleepForMSec(75);
			continue; // let's loop again without locking everything for long time
		 }
		// task to be executed, remove from queue (we may reinsert again later, cos of sorting)
		this->tasks.erase(this->tasks.begin(), false);
		std::sort(this->tasks.begin(), this->tasks.end(), SchedulerTaskSortPredicate);
		this->tasks.transactionEnd();
		if ( this->workersPool->getThreadCount() == 0 ) {
			task_executing(task, NULL);
		 }
		else {
			this->workersPool->getThreads().transactionBeginRead();
			SchedulerTaskExecuter* bTw = this->workersPool->waitNextIdle_(10000);
			if ( bTw == NULL ) {
				Log(LOG_WARNING, " Task#%"PRId64" : Threads seems Locked [after 10sec timeout]!", task->getID() );
				bTw = this->workersPool->waitNextIdle_(-1);
			 }
			Log(LOG_INFO, " Task#%"PRId64"", task->getID() );
			bTw->setTask(task);
			this->workersPool->getThreads().transactionEnd();
		 }
	 }
	Log(LOG_DEBUG, ":runThread(): EXITING" );
 }

void Scheduler::exitingThread() {
	this->workersPool->setWorkersCount<Scheduler>(0, this);
 }

////////////////////////////////////////////////////////////////////////////////////////////////// RUN THREAD
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////// TASK MANAGEMENT

WRESULT Scheduler::scheduleTask( SchedulerTask* newTask ) {
	if ( newTask == NULL ) return WRET_ERR_PARAM;
	if ( newTask->getTimeTick() == 0 ) {
		Log(LOG_CRITICAL, ":scheduleTask(%"PRId64",%"PRId64") Invalid TimeTick..", newTask->getID(), newTask->getTimeTick() );
		return WRET_ERR_ILLEGAL_USE;
	 }
	if ( newTask->getTimeTick() - currentCronTime() < 0 ) { // we are late!
		Log(LOG_CRITICAL, ":scheduleTask(%"PRId64",%"PRId64") Invalid TimeTick (%"PRId64"s in the past)..", newTask->getID(), newTask->getTimeTick(), currentCronTime()-newTask->getTimeTick() );
		return WRET_ERR_ILLEGAL_USE;
	 }
	Log(LOG_INFO, ":scheduleTask(%"PRId64",%"PRId64") Adding..", newTask->getID(), newTask->getTimeTick() );
	this->tasks.transactionBeginWrite();
	this->tasks.push_back( newTask ); // timetick/priority ordered FIFO
	std::sort(this->tasks.begin(), this->tasks.end(), SchedulerTaskSortPredicate);
	this->tasks.transactionEnd();
	return WRET_OK;
 }

WRESULT Scheduler::setupTask( int64 uid, int64 next_interrupt ) {
	if ( next_interrupt - currentCronTime() < 0 ) { // we are late!
		Log(LOG_CRITICAL, ":scheduleTask(%"PRId64",%"PRId64") Invalid TimeTick (%"PRId64"s in the past)..",  uid, next_interrupt, currentCronTime()-next_interrupt );
		return WRET_ERR_ILLEGAL_USE;
	 }
	this->tasks.transactionBeginWrite();
	tSchedulerTaskList::Iterator it;
	const tSchedulerTaskList::Iterator it_end = this->tasks.end();
	for ( it=this->tasks.begin(); it!=it_end; ++it ) {
		if ( *it == NULL ) continue;
		if ( (*it)->getID() != uid ) continue;
		if ( next_interrupt == 0 ) {
			Log(LOG_INFO, ":setupTask(%"PRId64") Removing Task (previous tick @ %"PRId64")..", uid, (*it)->getTimeTick() );
			this->tasks.erase(it, true);
		 }
		else {
			Log(LOG_INFO, ":setupTask(%"PRId64",%"PRId64") Updating Task..", uid, next_interrupt );
			(*it)->setTimeTick( next_interrupt );
		 }
		std::sort(this->tasks.begin(), this->tasks.end());
		this->tasks.transactionEnd();
		return WRET_OK;
	 }
	Log(LOG_INFO, ":setupTask(%"PRId64",%"PRId64") Adding Task..", uid, next_interrupt );
	SchedulerTask* newTask = new SchedulerTask();
	newTask->setID( uid );
	newTask->setTimeTick( next_interrupt );
	this->tasks.push_back(newTask);
	std::sort(this->tasks.begin(), this->tasks.end(), SchedulerTaskSortPredicate);
	this->tasks.transactionEnd();
	return WRET_OK;
 }

void Scheduler::task_executing( SchedulerTask* task, SchedulerTaskExecuter* ) {
	if ( task == NULL ) return;
	Log(LOG_VERBOSE, ":task_executing(%"PRId64")", task->getID() );
	int64 now;
	// let's wait till it's right moment
	while(this->running) {
		now = currentCronTime();
		if ( task->getTimeTick() <= now )
			break;
	 }
	if ( this->listener != NULL )
		this->listener->taskEvent(task, now, this);
	delete task;
 }

///////////////////////////////////////////////////////////////////////////////////////////// TASK MANAGEMENT
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////// SETTERS

int64 Scheduler::currentCronTime() const {
	time_t rawtime;
	struct tm* timeinfo;
	time ( &rawtime );
	timeinfo = localtime ( &rawtime );
	rawtime = mktime( timeinfo );
	return (int64)rawtime;
 }

WRESULT Scheduler::setListener( ISchedulerListener* ptr ) {
	if ( isThreadRunning() ) return WRET_ERR_ILLEGAL_USE;
	this->listener = ptr;
	return WRET_OK;
 }

void Scheduler::setDisabled( bool value ) {
	MutexLockerWrite mL(&this->configMux);
	this->disabled = value;
 }

///////////////////////////////////////////////////////////////////////////////////////////////////// SETTERS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
///////////////////////////////////////////////////////////////////////////////////////////////////// GETTERS

bool Scheduler::isDisabled() const {
	MutexLockerRead mL(&this->configMux);
	return this->disabled;
}

///////////////////////////////////////////////////////////////////////////////////////////////////// GETTERS
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
/////////////////////////////////////////////////////////////////////////////////////////////// SchedulerTask

 WOSH_REGISTER(wosh::xtended::SchedulerTask, "wosh::ISerializable", 1.01, _static_SchedulerTask )

bool SchedulerTask::updateHash( DataHashContext* hash_context ) const {
	if ( hash_context == NULL ) return false;
	bool ret = true;
	ret = hash_context->update( this->id ) & ret;
	ret = hash_context->update( this->timeTick ) & ret;
	ret = hash_context->update( this->priority ) & ret;
	return ret;
 }

std::string SchedulerTask::toString() const {
	return Utilities::format("SchedulerTask#%"PRId64, this->id);
 }

DataFieldCollector<SchedulerTask>* SchedulerTask::getDataFields() {
	DataFieldCollector<SchedulerTask>* fields = new DataFieldCollector<SchedulerTask>();
	fields->add<int64, Variant>(&SchedulerTask::getID, "ID", 0);
	fields->add<int64, Variant>(&SchedulerTask::getTimeTick, "TimeTick", 1);
	fields->add<long, Variant>(&SchedulerTask::getPriority, "Priority", 2);
	return fields;
 }

/////////////////////////////////////////////////////////////////////////////////////////////// SchedulerTask
//////////////////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////////////////

void SchedulerTaskExecuter::crashDetected( double probability ) {
	this->parent->Log(LOG_CRITICAL, ":crashDetected(%f)", probability );
 }

void SchedulerTaskExecuter::setTask( SchedulerTask* task_ptr ) {
	MutexLockerWrite mL(&this->processingLock);
	this->task = task_ptr;
 }

void SchedulerTaskExecuter::work() {
	this->processingLock.lockForRead();
	if ( this->task == NULL ) {
		this->processingLock.unLock();
		ThreadImpl::sleepForMSec(50);
		setThreadAlive();
		return;
	 }
	this->processing = true;
	setThreadAlive();
	this->parent->task_executing(this->task, this);
	this->task = NULL;
	this->processing = false;
	this->processingLock.unLock();
 }

//////////////////////////////////////////////////////////////////////////////////////////////////////////////

 }; // namespace xtended
}; // namespace wosh
